package com.shujia.window

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.{EventTimeSessionWindows, ProcessingTimeSessionWindows}
import org.apache.flink.streaming.api.windowing.time.Time

object Demo2SessionWindow {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val linesDS: DataStream[String] = env.socketTextStream("master", 9999)
    /*

001,1637893735000
001,1637893736000
001,1637893741000
001,1637893745000
001,1637893745000
001,1637893746000
001,1637893747000
001,1637893750000
001,1637893752000
001,1637893756000
     */
    val kvDS: DataStream[(String, Long)] = linesDS.map(line => {
      val split: Array[String] = line.split(",")
      (split(0), split(1).toLong)
    })
      //指定时间字段
      .assignAscendingTimestamps(_._2)

    /**
      * 多久没有数据产生开始计算，每一个key是独立
      *
      * ProcessingTimeSessionWindows： 处理时间的会话窗口
      * EventTimeSessionWindows ： 事件时间的会话窗口
      *
      * 时间时间的会话窗口需要新的数据来触发
      */
    kvDS
      .map(kv => (kv._1, 1))
      .keyBy(_._1)
      .window(EventTimeSessionWindows.withGap(Time.seconds(5)))
      .sum(1)
      .print()


    env.execute()

  }

}
